package com.whiteseason.spark.streaming

import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object FlumeWordCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("flumeWordCount")
    val context = new SparkContext(conf)
    context.setLogLevel("WARN")
    //获取上下文对象
    val ssc = new StreamingContext(context , Seconds(3))

    val lines = FlumeUtils.createStream(ssc, "192.168.1.115", 8888)
    val words = lines.map(x => new String(x.event.getBody.array()).trim)
      .flatMap(_.split(" "))
    val pairs = words.map(word => (word,1))
    val wordCounts = pairs.reduceByKey(_ + _)
    wordCounts.print()

    ssc.start()
    ssc.awaitTermination()
  }

}
